#include "../pubsubserver_better.h"

using namespace pubsub;

PubSubServerBetter::PubSubServerBetter(EventLoop *loop, const InetAddress &listenAddr)
    : _loop(loop),
      _server(loop, listenAddr)
{
    _server.setConnectionCallback(std::bind(&PubSubServerBetter::onConnection, this, _1));
    _server.setMessageCallback(std::bind(&PubSubServerBetter::onMessage, this, _1, _2, _3));
    _loop->runEvery(2.0, std::bind(&PubSubServerBetter::timePublish, this));
}
void PubSubServerBetter::setThreadNumber(int numthread)
{
    _server.setThreadNum(numthread);
}

void PubSubServerBetter::start()
{
    _server.setThreadInitCallback(std::bind(&PubSubServerBetter::threadInit, this, _1));
    _server.start();
}

void PubSubServerBetter::threadInit(EventLoop *loop)
{
    assert(LocalTopics::pointer() == NULL);
    LocalTopics::instance(); // 1.在线程里面初始化ConnectionList
    assert(LocalTopics::pointer() != NULL);

    MutexLockGuard lock(_mutex); //加锁
    _loops.insert(loop);         // 2.提取EventLoop
}

void PubSubServerBetter::onConnection(const TcpConnectionPtr &conn)
{
    if (conn->connected())
    {
        conn->setContext(ConnectionSubscription());
    }
    else
    {
        const ConnectionSubscription &connSub =
            boost::any_cast<const ConnectionSubscription &>(conn->getContext());
        for (auto it = connSub.begin(); it != connSub.end();)
        {
            doUnSubscribe(conn, *it++);
        }
    }
}
void PubSubServerBetter::onMessage(const TcpConnectionPtr &conn,
                                   Buffer *buff,
                                   Timestamp receiveTime)
{
    ParseResult result = kSuccess;
    while (result == kSuccess)
    {
        std::string cmd;
        std::string topic;
        std::string content;
        result = parseMessage(buff, &cmd, &topic, &content);
        if (result == kSuccess)
        {
            if (cmd == "pub")
            {
                doPublish(conn->name(), topic, content, receiveTime);
            }
            else if (cmd == "sub")
            {
                LOG_INFO << conn->name() << " subscribes " << topic;
                doSubscribe(conn, topic);
            }
            else if (cmd == "unsub")
            {
                doUnSubscribe(conn, topic);
            }
            else
            {
                conn->shutdown();
                result = kError;
            }
        }
        else if (result == kError)
        {
            conn->shutdown();
        }
    }
}
void PubSubServerBetter::timePublish()
{
    Timestamp now = Timestamp::now();
    doPublish("internal", "utc_time", now.toFormattedString(), now);
}
void PubSubServerBetter::doSubscribe(const TcpConnectionPtr &conn,
                                     const std::string &topic)
{
    ConnectionSubscription *connSub = boost::any_cast<ConnectionSubscription>(conn->getMutableContext());
    LOG_DEBUG << " doSubscribe PID : " << CurrentThread::tid();
    connSub->insert(topic);
    getTopic(topic).add(conn);
}
void PubSubServerBetter::doUnSubscribe(const TcpConnectionPtr &conn,
                                       const std::string &topic)
{
    LOG_INFO << conn->name() << " unsubscribes " << topic;
    getTopic(topic).remove(conn);
    ConnectionSubscription *connSub = boost::any_cast<ConnectionSubscription>(conn->getMutableContext());
    connSub->erase(topic);
}
void PubSubServerBetter::doPublish(const std::string &source,
                                   const std::string &topic,
                                   const std::string &content,
                                   Timestamp time)
{
    // //放入queueloop()的函数都得std::bind成void()类型
    // EventLoop::Functor f = std::bind(&Topic::publish, &getTopic(topic), content,time);
    EventLoop::Functor f = std::bind(&PubSubServerBetter::distributeMessage, this, topic, content, time);
    LOG_DEBUG;

    int cout = 0;
    MutexLockGuard lock(_mutex); //加锁->_loops
    for (std::set<EventLoop *>::iterator it = _loops.begin();
         it != _loops.end();
         ++it)
    {
        (*it)->queueInLoop(f); //把每个eventLoop里面的conn都广播一遍
    }
    LOG_DEBUG;
}
void PubSubServerBetter::distributeMessage(const std::string &topic,
                                           const std::string &content,
                                           Timestamp time)
{
    getTopic(topic).publish(content, time);
}

Topic &PubSubServerBetter::getTopic(const std::string &topic)
{
    LOG_TRACE << " PID : " << CurrentThread::tid();
    auto it = LocalTopics::instance().find(topic);
    if (it == LocalTopics::instance().end())
    {
        it = LocalTopics::instance().insert(std::make_pair(topic, Topic(topic))).first;
    }
    LOG_DEBUG << " topic name: " << it->first;
    return it->second;
}
